package xin.marcher.module.inventory.manage;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import xin.marcher.module.common.tair.TairCache;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 内部处理缓存消息使用的队列
 */
@Component
public class CalculateThresholdQueue {

    /**
     * 处理内部消息的队列
     */
    private final List<BlockingQueue> calculateQueue = new ArrayList<>();

    @Resource
    private TairCache tairCache;

    /**
     * 下一个任务处理队列在列表中的下标
     */
    private AtomicInteger index = new AtomicInteger();

    /**
     * 配置的消息内存队列数量
     */
    @Value("${calculate.threshold-num:32}")
    private Integer thresholdQueueNum;

    @PostConstruct
    public void init() {
        ExecutorService executors = Executors.newFixedThreadPool(thresholdQueueNum);
        for (int i = 0; i < thresholdQueueNum; i++) {
            //设置一个队列最大容纳数量
            BlockingQueue blockingQueue = new ArrayBlockingQueue(150000);
            calculateQueue.add(blockingQueue);
            executors.execute(new CalculateThresholdRunner(blockingQueue, tairCache));
        }
    }

    /**
     * 轮询
     *
     * @param object
     * @return
     */
    public boolean offerByRoundRobin(Object object) {
        index.compareAndSet(thresholdQueueNum * 10000, 0);
        boolean offer = calculateQueue.get(index.getAndIncrement() % thresholdQueueNum).offer(object);
        return offer;
    }
}
